package ddbs;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Map;

/*
 *MSG Type :<version><type><id><status><len><content>
 * For example v1 1 +-23 200 1 "x" 
 * Type:workload
 * id   + for quesry
 *      - for response
 * 
 * */
public class Peer {
    private int id;
    public SocketChannel toPeerChannel;
    private Boolean connected;
    public PeerMgr manager;
    public String serverIp;
    public int serverPort;
    Map<Integer, ProxyWork> pwmap;
    int transferSizeS = 0;
    int transferSizeR = 0;
    public Peer(int id) {
        this.pwmap = new HashMap<Integer, ProxyWork>();
        this.id = id;
        connected = false;
    }

    public Boolean getConnected() {
        return connected;
    }

    synchronized public Boolean setConnected(Boolean connected) {
        if (!connected) {
            this.connected = connected;
            return true;
        }
        if (this.connected)
            return false;
        this.connected = connected;
        return true;
    }

    public int getId() {
        return this.id;
    }

    public static byte[] int2Byte(int intValue) {
        byte[] b = new byte[4];
        for (int i = 0; i < 4; i++) {
            b[i] = (byte) (intValue >> 8 * (3 - i) & 0xFF);
            // System.out.print(Integer.toBinaryString(b[i])+" ");
            // System.out.print((b[i]& 0xFF)+" ");
        }
        return b;
    }

    public static int byte2Int(byte[] b) {
        int intValue = 0;
        for (int i = 0; i < b.length; i++) {
            intValue += (b[i] & 0xFF) << (8 * (3 - i));
            // System.out.print(Integer.toBinaryString(intValue)+" ");
        }
        return intValue;
    }

    synchronized public Boolean sendMsg(MSG msg) {
        if (!this.getConnected()) {
            System.out.println(" Failed to Send message to peer " + this.id+" while it is not connected");
            return false;
        }
        try {
            SocketChannel channel = this.toPeerChannel;
            String contents = msg.type + "!" + msg.status + "!" + msg.id + "!"
                    + msg.len + "!" + msg.content;
            int length = contents.length();
            transferSizeS +=length;
            ByteBuffer bf = ByteBuffer.allocate(4+length);
            bf.clear();
            bf.put(int2Byte(length));
            bf.put(contents.getBytes());
            bf.flip();
            while (bf.hasRemaining()) {
                channel.write(bf);
            }
            if (length > 100)        length = 100;
            System.out.println("send message to " + this.id + " :" + contents.substring(0,  length));
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    // Communication
    public boolean IsSelf() {
        return this == manager.getSelf();
    }

    public void sendWork(LocalWork lWork) {
        MSG msg = lWork.toMSG();
        if (!sendMsg(msg)) {
            lWork.sendWorkFailed();
        } else {
            manager.sch.addWork(lWork);
        }
    }

    public void sendWorkResult(ProxyWork pw) {
        MSG msg = pw.toMSG();
        sendMsg(msg);
    }

    public void addProxyWork(ProxyWork pw) {
        pwmap.put(pw.getId(), pw);
    }

    public void rmProxyWork(ProxyWork pw) {
        pwmap.remove(pw.getId());
    }
}
